#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Time          : 2024/4/26 10:56
# File          : kafka_consumer.py
# @Author       : MingTai
# @Department   : automation_qa
from kafka import KafkaConsumer

from app.business.config.config_base import dev_kafka_server_list_common


def kafka_consumer_example(topic_name, kafka_servers=dev_kafka_server_list_common):
    # 需要引入包：
    # python-snappy
    # kafka - python

    """
    消费指定Topic的消息
    :param topic_name: Kafka主题名
    :param kafka_servers: Kafka服务器地址列表
    """
    # 创建一个KafkaConsumer实例
    consumer = KafkaConsumer(
        topic_name,
        bootstrap_servers=kafka_servers,
        auto_offset_reset='earliest',  # 从最早的消息开始消费
        group_id='my-group'  # 消费者组ID
    )

    print(f"Waiting for messages in topic: {topic_name}...")
    # 无限循环，消费消息
    for message in consumer:
        # 解码消息内容
        print(f"Received message: {message.value.decode('utf-8')}")


# 使用示例
if __name__ == '__main__':
    TOPIC_NAME = 'test_topic'  # 替换为你的Kafka Topic
    kafka_consumer_example(TOPIC_NAME)
